package com.lee.windowFunction;

import com.lee.*;
import com.lee.entity.SessionResult;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * Created by liruopeng on 2018/1/18.
 */

public class TCPreorganizetionFunction implements WindowFunction<networkData,SessionResult,String,TimeWindow> {

    private Logger logger = LoggerFactory.getLogger(TCPreorganizetionFunction.class);

    @Override
    public void apply(String key, TimeWindow timeWindow, Iterable<networkData> iterable, Collector<SessionResult> collector) throws Exception {
        SessionResult sr = new SessionResult();
        networkData data = iterable.iterator().next();
        Ethernet eth = (Ethernet) data.getNetworkAccessLayer();
        TCP tcp = (TCP) data.getTransportLayer();
        IP ip = (IP)data.getInternetLayer();
        if(!isInner(ip.getDstIP().toString())){
            sr.setSrcIP(ip.getSrcIP().toString());
            sr.setSrcPort(tcp.getSrcPort());
            sr.setDstIP(ip.getDstIP().toString());
            sr.setDstPort(tcp.getDstPort());
            sr.setStartTime(eth.getTime());
            sr.setProtocol(tcp.getApplicationProtocol().toString());
        }else {
            sr.setSrcIP(ip.getDstIP().toString());
            sr.setSrcPort(tcp.getDstPort());
            sr.setDstIP(ip.getSrcIP().toString());
            sr.setDstPort(tcp.getSrcPort());
            sr.setStartTime(eth.getTime());
            sr.setProtocol(tcp.getApplicationProtocol().toString());
        }
        for(networkData dataIter: iterable){
            eth = (Ethernet) dataIter.getNetworkAccessLayer();
            tcp = (TCP) dataIter.getTransportLayer();
            ip = (IP)dataIter.getInternetLayer();
            if(ip.getSrcIP().toString().compareTo(sr.getSrcIP())==0){//说明是发给外网的包
                sr.setSendLen(sr.getSendLen()+eth.getLength());
            }else if(ip.getDstIP().toString().compareTo(sr.getSrcIP())==0){//说明是接受到外网的包
                sr.setRecLen(sr.getSendLen()+eth.getLength());
            }else{//说明不是这个会话的数据，前面计算出错了
                logger.error("你的窗口中出现了不属于该窗口的数据包，请尽快处理。");
            }
            sr.setEndTime(eth.getTime());

        }
        collector.collect(sr);
    }

    /*
* 判断是否是内网IP
* */
    public  boolean isInner(String ip)
    {
        String reg = "(10|172|192)\\.([0-1][0-9]{0,2}|[2][0-5]{0,2}|[3-9][0-9]{0,1})\\.([0-1][0-9]{0,2}|[2][0-5]{0,2}|[3-9][0-9]{0,1})\\.([0-1][0-9]{0,2}|[2][0-5]{0,2}|[3-9][0-9]{0,1})";//正则表达式=。 =、懒得做文字处理了、
        Pattern p = Pattern.compile(reg);
        Matcher matcher = p.matcher(ip);
        return matcher.find();
    }
}